package flink;

import bean.socketBean;
import com.alibaba.fastjson.JSONObject;
import function.TopNProcesstionFuncton;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.*;


public class TopNCompute {
    public static void main(String[] args) throws Exception{
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //获取socket输入流
        DataStreamSource<String> dataStreamSource = env.socketTextStream("192.168.1.10", 7000);
        //提取watermark
        SingleOutputStreamOperator<socketBean> sockerDs = dataStreamSource.map(JSONObject::parseObject).map(lines -> {
            socketBean socketBean = new socketBean();
            socketBean.setId(lines.getString("id"));
            socketBean.setMoney(lines.getDouble("money"));
            socketBean.setTime(lines.getString("time"));
            return socketBean;
        }).assignTimestampsAndWatermarks(WatermarkStrategy.<socketBean>forMonotonousTimestamps().withTimestampAssigner(new SerializableTimestampAssigner<socketBean>() {
                    @Override
                    public long extractTimestamp(socketBean element, long recordTimestamp) {
                        return Long.valueOf(element.getTime());
                    }
                })
        );

/*        //计算10分钟内金额最高的前三名统计所有的key都被聚合在一个窗口内了
        SingleOutputStreamOperator<socketBean> process = sockerDs
                .windowAll(TumblingEventTimeWindows.of(Time.minutes(10))).process(new TopNProcesstionFuncton());*/

        //按id计算10分钟内每个id最大的金额前三名
        SingleOutputStreamOperator<socketBean> process = sockerDs.keyBy(new KeySelector<socketBean, String>() {
            @Override
            public String getKey(socketBean value) throws Exception {
                return value.getId();
            }
        }).window(TumblingEventTimeWindows.of(Time.minutes(10))).apply(new WindowFunction<socketBean, socketBean, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<socketBean> input, Collector<socketBean> out) throws Exception {
                System.out.println(">>>>>>>>>>>>>>>"+s);
                List<socketBean> l = new ArrayList<socketBean>();
                Iterator<socketBean> iterator = input.iterator();
                while (iterator.hasNext()){
                    socketBean socketBean = iterator.next();
                    l.add(socketBean);
                }
                l.sort(Comparator.comparing(socketBean::getMoney).reversed());
                for (int i=0;i<l.size();i++){
                    if(l.get(i).getId().equals(s)){
                        if(i<3){
                            out.collect(l.get(i));
                        }
                    }
                }
            }
        });

        //打印数据
        process.print();

        //启动程序
        env.execute("TopNCompute");
    }
}
